package com.qianbao.finance.sync.modules.elasticsearch.scheduling;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.Message;

import com.qianbao.finance.sync.modules.elasticsearch.client.CanalClientFactory;
import com.qianbao.finance.sync.modules.elasticsearch.event.DeleteCanalEvent;
import com.qianbao.finance.sync.modules.elasticsearch.event.InsertCanalEvent;
import com.qianbao.finance.sync.modules.elasticsearch.event.UpdateCanalEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;


@Component
public class CanalScheduling implements Runnable, ApplicationContextAware {
    private static final Logger logger = LoggerFactory.getLogger(CanalScheduling.class);
    private ApplicationContext applicationContext;


    @Value("${canal.destination}")
    private String canalDestination;

    @Resource
    private CanalClientFactory canalClientFactory;
    @Scheduled(fixedDelay = 2000)
    @Override
    public void run() {
        String[] canalDestinations = canalDestination.split(",");
        for(String canalDestination : canalDestinations){
                    logger.info("同步："+canalDestination+"数据");
                    CanalConnector canalConnector = canalClientFactory.getCanalConnector(canalDestination);
                     try {
                         int batchSize = 1000;
                         Message message = canalConnector.getWithoutAck(batchSize);
                         long batchId = message.getId();
                         logger.info("CanalScheduling scheduled_batchId=" + batchId+"  canalDestination:"+canalDestination);
                         try {
                             List<Entry> entries = message.getEntries();
                             if (batchId != -1 && entries.size() > 0) {
                                 entries.forEach(entry -> {
                                     if (entry.getEntryType() == EntryType.ROWDATA) {
                                         publishCanalEvent(entry);
                                     }
                                 });
                             }
                             canalConnector.ack(batchId);
                         } catch (Exception e) {
                             logger.error("发送监听事件失败！batchId回滚,batchId=" + batchId, e);
                             canalConnector.rollback(batchId);
                         }
                     } catch (Exception e) {
                         try{
                             canalClientFactory.refresh(canalDestination);
                         }catch (Exception e1){
                             logger.error("连接canal异常！", e);
                         }
                         logger.error("canal_scheduled异常！", e);
                     }
        }
    }

    private void publishCanalEvent(Entry entry) {
        EventType eventType = entry.getHeader().getEventType();
        logger.info("publishCanalEvent type=" + eventType);
        switch (eventType) {
            case INSERT:
                applicationContext.publishEvent(new InsertCanalEvent(entry));
                break;
            case UPDATE:
                applicationContext.publishEvent(new UpdateCanalEvent(entry));
                break;
            case DELETE:
                applicationContext.publishEvent(new DeleteCanalEvent(entry));
                break;
            default:
                break;
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
